<!DOCTYPE html>
<html xmlns="http://www.w3.org/1999/xhtml">
  <head>

    <meta http-equiv="Content-Type" content="text/html; charset=utf-8" />
    <meta content="Cask Data, Inc." name="author" />
<meta content="Copyright © 2015-2016 Cask Data, Inc." name="copyright" />


    <meta name="git_release" content="6.1.1">
    <meta name="git_hash" content="05fbac36f9f7aadeb44f5728cea35136dbc243e5">
    <meta name="git_timestamp" content="2020-02-09 08:22:47 +0800">
    <title>Batch Data Processing with CDAP using Workflow</title>

    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />
    <link rel="stylesheet" href="../_static/pygments.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/bootstrap-3.3.6/css/bootstrap-theme.min.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/bootstrap-sphinx.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-dynamicscrollspy-4.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/cdap-jquery.mCustomScrollbar.css" type="text/css" />
    <link rel="stylesheet" href="../_static/css/abixTreeList-2.css" type="text/css" />
    <link rel="stylesheet" href="../_static/cdap-bootstrap.css" type="text/css" />

    <script type="text/javascript">
      var DOCUMENTATION_OPTIONS = {
        URL_ROOT:    '',
        VERSION:     '6.1.1',
        COLLAPSE_INDEX: false,
        FILE_SUFFIX: '.html',
        HAS_SOURCE:  false
      };
    </script>
    <script type="text/javascript" src="../_static/jquery.js"></script>
    <script type="text/javascript" src="../_static/underscore.js"></script>
    <script type="text/javascript" src="../_static/doctools.js"></script>
    <script type="text/javascript" src="../_static/language_data.js"></script>

    <link rel="shortcut icon" href="../_static/favicon.ico"/>
    <link rel="index" title="Index" href="../genindex.html" />
    <link rel="search" title="Search" href="../search.html" />
    <link rel="top" title="Cask Data Application Platform 6.1.1 Documentation" href="../index.html" />
    <link rel="up" title="How-To Guides" href="index.html" />
    <link rel="prev" title="Consuming Twitter Data in Real Time" href="cdap-twitter-ingest-guide.html" />
    <!-- block extrahead -->
    <meta charset='utf-8'>
    <meta http-equiv='X-UA-Compatible' content='IE=edge,chrome=1'>
    <meta name='viewport' content='width=device-width, initial-scale=1.0, maximum-scale=1'>
    <meta name="apple-mobile-web-app-capable" content="yes">
    <!-- block extrahead end -->

</head>
<body role="document">

<!-- block navbar -->
<div id="navbar" class="navbar navbar-inverse navbar-default navbar-fixed-top">
    <div class="container-fluid">
      <div class="row">
        <div class="navbar-header">
          <!-- .btn-navbar is used as the toggle for collapsed navbar content -->
          <a class="navbar-brand" href="../table-of-contents/../../index.html">
            <span><img alt="CDAP logo" src="../_static/cdap_logo.svg"/></span>
          </a>

          <button type="button" class="navbar-toggle" data-toggle="collapse" data-target=".nav-collapse">
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
            <span class="icon-bar"></span>
          </button>

          <div class="pull-right">
            <div class="dropdown version-dropdown">
              <a href="#" class="dropdown-toggle" data-toggle="dropdown"
                role="button" aria-haspopup="true" aria-expanded="false">
                v 6.1.1 <span class="caret"></span>
              </a>
              <ul class="dropdown-menu">
                <li><a href="//docs.cdap.io/cdap/5.1.2/en/index.html">v 5.1.2</a></li>
                <li><a href="//docs.cdap.io/cdap/4.3.4/en/index.html">v 4.3.4</a></li>
              </ul>
            </div>
          </div>
          <form class="navbar-form navbar-right navbar-search" action="../search.html" method="get">
            <div class="form-group">
              <div class="navbar-search-image material-icons"></div>
              <input type="text" name="q" class="form-control" placeholder="  Search" />
            </div>
            <input type="hidden" name="check_keywords" value="yes" />
            <input type="hidden" name="area" value="default" />
          </form>

          <div class="collapse navbar-collapse nav-collapse navbar-right navbar-navigation">
            <ul class="nav navbar-nav"><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../index.html">简介</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link current" href="../table-of-contents/../../guides.html">手册</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../reference-manual/index.html">参考</a></li><li class="docsite-nav-tab-container"><a class="docsite-nav-tab-link " href="../table-of-contents/../../faqs/index.html">帮助</a></li>
            </ul>
          </div>

        </div>
      </div>
    </div>
  </div><!-- block navbar end -->
<!-- block main content -->
<div class="main-container container">
  <div class="row"><div class="col-md-2">
      <div id="sidebar" class="bs-sidenav scrollable-y-outside" role="complementary">
<!-- theme_manual: examples-manual -->
<!-- theme_manual_highlight: guides -->
<!-- sidebar_title_link: ../table-of-contents/../../guides.html -->

  <div role="note" aria-label="manuals links"><h3><a href="../table-of-contents/../../guides.html">Guides</a></h3>

    <ul class="this-page-menu">
      <li class="toctree-l1"><a href="../table-of-contents/../../user-guide/index.html" rel="nofollow">用户手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../developer-manual/index.html" rel="nofollow">开发手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../admin-manual/index.html" rel="nofollow">管理手册</a>
      </li>
      <li class="toctree-l1"><a href="../table-of-contents/../../integrations/index.html" rel="nofollow">集成手册</a>
      </li>
      <li class="toctree-l1"><b><a href="../table-of-contents/../../examples-manual/index.html" rel="nofollow">最佳实践</a></b>
      <nav class="pagenav">
      <ul class="current">
<li class="toctree-l1"><a class="reference internal" href="../index.html">Introduction</a></li>
<li class="toctree-l1 current"><a class="reference internal" href="index.html">How-To Guides</a><ul class="current">
<li class="toctree-l2"><a class="reference internal" href="cdap-bi-guide.html">Analyzing CDAP Data from BI Tools</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-cube-guide.html">Data Analysis with OLAP Cube</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-etl-guide.html">Creating Hydrator Applications using CDAP System Artifacts</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-flow-guide.html">Real-Time Data Processing with a Flow</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-flume-guide.html">Ingesting Data into CDAP using Apache Flume</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-kafka-ingest-guide.html">Consuming Data from Kafka</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-mapreduce-guide.html">Batch Data Processing with CDAP</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-spark-guide.html">Iterative Data Processing with Apache Spark</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-timeseries-guide.html">Storing Timeseries Data</a></li>
<li class="toctree-l2"><a class="reference internal" href="cdap-twitter-ingest-guide.html">Consuming Twitter Data in Real Time</a></li>
<li class="toctree-l2 current"><a class="current reference internal" href="#">Batch Data Processing with CDAP using Workflow</a></li>
</ul>
</li>
</ul>
</nav>
      </li>
    </ul>
  </div></div>
    </div><div class="col-md-8 content" id="main-content">
    
  <div class="section" id="batch-data-processing-with-cdap-using-workflow">
<span id="cdap-workflow-guide"></span><h1>Batch Data Processing with CDAP using Workflow<a class="headerlink" href="#batch-data-processing-with-cdap-using-workflow" title="Permalink to this headline">🔗</a></h1>
<blockquote class="pull-quote">
<div><strong>Source Code Repository:</strong> Source code (and other resources) for this guide are available at the
<a class="reference external" href="https://github.com/cdap-guides/cdap-workflow-guide/tree/develop">CDAP Guides GitHub repository</a>.</div></blockquote>
<p>The Workflow system in <a class="reference external" href="http://cdap.io">Cask Data Application Platform (CDAP)</a>
allows specifying, executing, scheduling, and monitoring complex series of jobs
and tasks. In this guide, you will learn how it can be used to execute the
<a class="reference external" href="http://research.google.com/archive/mapreduce.html">MapReduce</a>
programs in parallel based on the evaluation of conditions.</p>
<div class="section" id="what-you-will-build">
<h2>What You Will Build<a class="headerlink" href="#what-you-will-build" title="Permalink to this headline">🔗</a></h2>
<p>This guide will take you through building a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/applications.html">CDAP application</a>
that uses ingested raw purchase events (of the form ‘<code class="docutils literal notranslate"><span class="pre">&lt;name&gt;</span> <span class="pre">bought</span> <span class="pre">&lt;n&gt;</span> <span class="pre">&lt;item&gt;s</span> <span class="pre">for</span> <span class="pre">$&lt;price&gt;</span></code>’, which are parsed
using a primitive parser for sentences) to compute in parallel the total purchases made by each customer along with
the total purchases made for each product.</p>
<p>You will:</p>
<ul class="simple">
<li>Build <code class="docutils literal notranslate"><span class="pre">PurchaseEventParser</span></code>, a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/mapreduce-programs.html">MapReduce program</a>
to parse the raw purchase events and create <code class="docutils literal notranslate"><span class="pre">Purchase</span></code> objects from them;</li>
<li>Build <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByCustomer</span></code>, a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/mapreduce-programs.html">MapReduce program</a>
to count the purchases made per customer;</li>
<li>Build <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByProduct</span></code>, a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/mapreduce-programs.html">MapReduce program</a>
to count the purchases made per product;</li>
<li>Build <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code>, a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/workflows.html">Workflow</a>
which will first execute the MapReduce program <code class="docutils literal notranslate"><span class="pre">PurchaseEventParser</span></code>. If the predicate <code class="docutils literal notranslate"><span class="pre">PurchaseEventVerifier</span></code>,
which uses the MapReduce counters emitted by the PEP to determine data quality, evaluates to true, the workflow will
in parallel execute the MapReduce program <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByCustomer</span></code> and <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByProduct</span></code> otherwise,
it will execute the action <code class="docutils literal notranslate"><span class="pre">ProblemLogger</span></code>;</li>
<li>Use
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/datasets/index.html">Datasets</a>
to persist results of the MapReduce programs; and</li>
<li>Build a
<a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/services.html">Service</a>
to serve the results via HTTP.</li>
</ul>
</div>
<div class="section" id="what-you-will-need">
<h2>What You Will Need<a class="headerlink" href="#what-you-will-need" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li><a class="reference external" href="http://www.oracle.com/technetwork/java/javase/downloads/index.html">JDK 7 or 8</a></li>
<li><a class="reference external" href="http://maven.apache.org/">Apache Maven 3.1+</a></li>
<li><a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/getting-started/standalone/index.html">CDAP SDK</a></li>
</ul>
</div>
<div class="section" id="lets-build-it">
<h2>Let’s Build It!<a class="headerlink" href="#lets-build-it" title="Permalink to this headline">🔗</a></h2>
<p>The following sections will guide you through building an application from scratch. If you
are interested in deploying and running the application right away, you can clone its
source code from this GitHub repository. In that case, feel free to skip the next two
sections and jump right to the
<a class="reference external" href="#build-and-run-application">Build and Run Application</a> section.</p>
<div class="section" id="application-design">
<h3>Application Design<a class="headerlink" href="#application-design" title="Permalink to this headline">🔗</a></h3>
<p>The application will assume that the purchase events are ingested
into a Stream. The events can be ingested into a Stream continuously
in real time or in batches; whichever way, it doesn’t affect the ability
of the MapReduce programs to consume them.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code> encapsulates the set of MapReduce programs, which extracts the required information from the
raw purchase events and computes the total purchases made by each customer and total purchases made for
each product in a specific time range. The results of the computation are persisted in Datasets.</p>
<p>Finally, the application contains a Service that exposes an HTTP endpoint to access the data stored in the Datasets.</p>
<a class="reference internal image-reference" href="../_images/app-design9.png"><img alt="../_images/app-design9.png" class="align-center" src="../_images/app-design9.png" style="width: 8in;" /></a>
</div>
<div class="section" id="implementation">
<h3>Implementation<a class="headerlink" href="#implementation" title="Permalink to this headline">🔗</a></h3>
<p>The first step is to construct our application structure. We will use a
standard Maven project structure for all of the source code files:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">./pom.xml</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/ProblemLogger.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/Purchase.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseCounterByProduct.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseCounterByCustomer.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseEventParser.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseEventVerifier.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseResultService.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseWorkflow.java</span>
<span class="go">./src/main/java/co/cdap/guides/workflow/PurchaseWorkflowApp.java</span>
</pre></div>
</div>
<p>The CDAP application is identified by the <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflowApp</span></code> class. This
class extends an <a class="reference external" href="http://docs.cdap.io/cdap/current/en/reference-manual/javadocs/co/cask/cdap/api/app/AbstractApplication.html">AbstractApplication</a>,
and overrides the <code class="docutils literal notranslate"><span class="pre">configure</span></code> method to define all of the application components:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">PurchaseWorkflowApp</span> <span class="kd">extends</span> <span class="n">AbstractApplication</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="s">&quot;PurchaseWorkflowApp&quot;</span><span class="p">);</span>
    <span class="n">setDescription</span><span class="p">(</span><span class="s">&quot;Application describing the Workflow&quot;</span><span class="p">);</span>

    <span class="n">addStream</span><span class="p">(</span><span class="k">new</span> <span class="n">Stream</span><span class="p">(</span><span class="s">&quot;purchaseEvents&quot;</span><span class="p">));</span>

    <span class="n">addMapReduce</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseEventParser</span><span class="p">());</span>
    <span class="n">addMapReduce</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseCounterByCustomer</span><span class="p">());</span>
    <span class="n">addMapReduce</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseCounterByProduct</span><span class="p">());</span>
    <span class="n">addWorkflow</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseWorkflow</span><span class="p">());</span>

    <span class="n">scheduleWorkflow</span><span class="p">(</span><span class="n">Schedules</span><span class="p">.</span><span class="na">builder</span><span class="p">(</span><span class="s">&quot;HourlySchedule&quot;</span><span class="p">)</span>
                       <span class="p">.</span><span class="na">setDescription</span><span class="p">(</span><span class="s">&quot;Schedule execution every 1 hour&quot;</span><span class="p">)</span>
                       <span class="p">.</span><span class="na">createTimeSchedule</span><span class="p">(</span><span class="s">&quot;0 * * * *&quot;</span><span class="p">),</span>
                     <span class="s">&quot;PurchaseWorkflow&quot;</span><span class="p">);</span>

    <span class="n">addService</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseResultService</span><span class="p">());</span>

    <span class="n">createDataset</span><span class="p">(</span><span class="s">&quot;purchaseRecords&quot;</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">createDataset</span><span class="p">(</span><span class="s">&quot;customerPurchases&quot;</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
    <span class="n">createDataset</span><span class="p">(</span><span class="s">&quot;productPurchases&quot;</span><span class="p">,</span> <span class="n">KeyValueTable</span><span class="p">.</span><span class="na">class</span><span class="p">);</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>The <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflowApp</span></code> application defines a new <a class="reference external" href="http://docs.cdap.io/cdap/current/en/developers-manual/building-blocks/streams.html">Stream</a>
where purchase events are ingested. Once the data is
ingested, the events can be processed in real time or batch. In our
application, we will process the events in batch using the
<code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code> program and compute the total purchases made by each customer
and the total purchases made for each product in a specific time range. We will use three MapReduce
programs <code class="docutils literal notranslate"><span class="pre">PurchaseEventParser</span></code>, <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByCustomer</span></code>, and <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByProduct</span></code> to apply
different processing on the purchase events and the Workflow <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code> to connect these MapReduce
programs.</p>
<p>The result of the Workflow execution is persisted into Datasets; the
application uses the <code class="docutils literal notranslate"><span class="pre">createDataset</span></code> method to define the Dataset. We use three datasets:
<code class="docutils literal notranslate"><span class="pre">purchaseRecords</span></code> to store the valid parsed purchase events; <code class="docutils literal notranslate"><span class="pre">customerPurchases</span></code> to store the total purchases
made by each customer; and <code class="docutils literal notranslate"><span class="pre">productPurchases</span></code> to store the total purchases made for each product.
The <code class="docutils literal notranslate"><span class="pre">Purchase</span></code> class defines the type used to store the parsed purchase events.</p>
<p>The application also adds a custom Workflow action <code class="docutils literal notranslate"><span class="pre">ProblemLogger</span></code>. When a Workflow executes a custom action,
it invokes the <code class="docutils literal notranslate"><span class="pre">run</span></code> method in the action. In <code class="docutils literal notranslate"><span class="pre">ProblemLogger</span></code>, we only add a log statement; however it could be
customized to send emails to the concerned parties.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code> is scheduled to execute every hour.</p>
<p>Finally, the application adds a service for querying the results from the Datasets.</p>
<p>Let’s take a closer look at the Workflow.</p>
<p>The <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code> extends an <a class="reference external" href="http://docs.cdap.io/cdap/current/en/reference-manual/javadocs/co/cask/cdap/api/workflow/AbstractWorkflow.html">AbstractWorkflow</a>
class and overrides the <code class="docutils literal notranslate"><span class="pre">configure</span></code> method:</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">PurchaseWorkflow</span> <span class="kd">extends</span> <span class="n">AbstractWorkflow</span> <span class="p">{</span>
  <span class="nd">@Override</span>
  <span class="kd">protected</span> <span class="kt">void</span> <span class="nf">configure</span><span class="p">()</span> <span class="p">{</span>
    <span class="n">setName</span><span class="p">(</span><span class="s">&quot;PurchaseWorkflow&quot;</span><span class="p">);</span>
    <span class="n">setDescription</span><span class="p">(</span><span class="s">&quot;Workflow to parse the purchase events and count the revenue per customer and per product&quot;</span><span class="p">);</span>

    <span class="n">addMapReduce</span><span class="p">(</span><span class="s">&quot;PurchaseEventParser&quot;</span><span class="p">);</span>

    <span class="n">condition</span><span class="p">(</span><span class="k">new</span> <span class="n">PurchaseEventVerifier</span><span class="p">())</span>
      <span class="p">.</span><span class="na">fork</span><span class="p">()</span>
        <span class="p">.</span><span class="na">addMapReduce</span><span class="p">(</span><span class="s">&quot;PurchaseCounterByCustomer&quot;</span><span class="p">)</span>
      <span class="p">.</span><span class="na">also</span><span class="p">()</span>
        <span class="p">.</span><span class="na">addMapReduce</span><span class="p">(</span><span class="s">&quot;PurchaseCounterByProduct&quot;</span><span class="p">)</span>
      <span class="p">.</span><span class="na">join</span><span class="p">()</span>
    <span class="p">.</span><span class="na">otherwise</span><span class="p">()</span>
      <span class="p">.</span><span class="na">addAction</span><span class="p">(</span><span class="k">new</span> <span class="n">ProblemLogger</span><span class="p">())</span>
    <span class="p">.</span><span class="na">end</span><span class="p">();</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p>In the <code class="docutils literal notranslate"><span class="pre">configure</span></code> method we specify the topology for connecting the programs which will run as a part of
the Workflow execution. As the first action in the <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflow</span></code>, we add the MapReduce program
<code class="docutils literal notranslate"><span class="pre">PurchaseEventParser</span></code>. This program will parse raw purchase events (using a primitive sentence parser) and
create <code class="docutils literal notranslate"><span class="pre">Purchase</span></code> objects from them.</p>
<p>After that, we add a <code class="docutils literal notranslate"><span class="pre">condition</span></code> in the Workflow, which takes a predicate <code class="docutils literal notranslate"><span class="pre">PurchaseEventVerifier</span></code>.
If the predicate evaluates to true, we <code class="docutils literal notranslate"><span class="pre">fork</span></code> the execution of the Workflow into two parallel branches.
One branch executes the <code class="docutils literal notranslate"><span class="pre">PurchaseCounterByCustomer</span></code> MapReduce program, while the other executes the
<code class="docutils literal notranslate"><span class="pre">PurchaseCounterByProduct</span></code> MapReduce program.</p>
<p>If the predicate evaluates to false, then actions in the <code class="docutils literal notranslate"><span class="pre">otherwise</span></code> section will be executed.
We have added a single custom action, <code class="docutils literal notranslate"><span class="pre">ProblemLogger</span></code> to the <code class="docutils literal notranslate"><span class="pre">otherwise</span></code> section as an example
of what is possible.</p>
<p>Lets take a closer look at the predicate <code class="docutils literal notranslate"><span class="pre">PurchaseEventVerifier</span></code>.</p>
<div class="highlight-java notranslate"><div class="highlight"><pre><span></span><span class="kd">public</span> <span class="kd">class</span> <span class="nc">PurchaseEventVerifier</span> <span class="kd">implements</span> <span class="n">Predicate</span><span class="o">&lt;</span><span class="n">WorkflowContext</span><span class="o">&gt;</span> <span class="p">{</span>

  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">TASK_COUNTER_GROUP_NAME</span> <span class="o">=</span> <span class="s">&quot;org.apache.hadoop.mapreduce.TaskCounter&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">MAP_INPUT_RECORDS_COUNTER_NAME</span> <span class="o">=</span> <span class="s">&quot;MAP_INPUT_RECORDS&quot;</span><span class="p">;</span>
  <span class="kd">private</span> <span class="kd">static</span> <span class="kd">final</span> <span class="n">String</span> <span class="n">MAP_OUTPUT_RECORDS_COUNTER_NAME</span> <span class="o">=</span> <span class="s">&quot;MAP_OUTPUT_RECORDS&quot;</span><span class="p">;</span>

  <span class="nd">@Override</span>
  <span class="kd">public</span> <span class="kt">boolean</span> <span class="nf">apply</span><span class="p">(</span><span class="n">WorkflowContext</span> <span class="n">workflowContext</span><span class="p">)</span> <span class="p">{</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">workflowContext</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">return</span> <span class="kc">false</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="n">WorkflowToken</span> <span class="n">token</span> <span class="o">=</span> <span class="n">workflowContext</span><span class="p">.</span><span class="na">getToken</span><span class="p">();</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">token</span> <span class="o">==</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="k">return</span> <span class="kc">false</span><span class="p">;</span>
    <span class="p">}</span>

    <span class="n">Value</span> <span class="n">mapInputRecords</span> <span class="o">=</span> <span class="n">token</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">TASK_COUNTER_GROUP_NAME</span> <span class="o">+</span> <span class="s">&quot;.&quot;</span> <span class="o">+</span> <span class="n">MAP_INPUT_RECORDS_COUNTER_NAME</span><span class="p">,</span>
                                      <span class="n">WorkflowToken</span><span class="p">.</span><span class="na">Scope</span><span class="p">.</span><span class="na">SYSTEM</span><span class="p">);</span>
    <span class="n">Value</span> <span class="n">mapOutputRecords</span> <span class="o">=</span> <span class="n">token</span><span class="p">.</span><span class="na">get</span><span class="p">(</span><span class="n">TASK_COUNTER_GROUP_NAME</span> <span class="o">+</span> <span class="s">&quot;.&quot;</span> <span class="o">+</span> <span class="n">MAP_OUTPUT_RECORDS_COUNTER_NAME</span><span class="p">,</span>
                                       <span class="n">WorkflowToken</span><span class="p">.</span><span class="na">Scope</span><span class="p">.</span><span class="na">SYSTEM</span><span class="p">);</span>
    <span class="k">if</span> <span class="p">(</span><span class="n">mapInputRecords</span> <span class="o">!=</span> <span class="kc">null</span> <span class="o">&amp;&amp;</span> <span class="n">mapOutputRecords</span> <span class="o">!=</span> <span class="kc">null</span><span class="p">)</span> <span class="p">{</span>
      <span class="c1">// Return true if at least 80% of the records were successfully parsed and emitted</span>
      <span class="c1">// by previous map job</span>
      <span class="k">return</span> <span class="p">(</span><span class="n">mapOutputRecords</span><span class="p">.</span><span class="na">getAsLong</span><span class="p">()</span> <span class="o">&gt;=</span> <span class="p">(</span><span class="n">mapInputRecords</span><span class="p">.</span><span class="na">getAsLong</span><span class="p">()</span> <span class="o">*</span> <span class="mi">80</span><span class="o">/</span><span class="mi">100</span><span class="p">));</span>
    <span class="p">}</span>
    <span class="k">return</span> <span class="kc">false</span><span class="p">;</span>
  <span class="p">}</span>
<span class="p">}</span>
</pre></div>
</div>
<p><code class="docutils literal notranslate"><span class="pre">PurchaseEventVerifier</span></code> needs to be a public class which implements the interface <code class="docutils literal notranslate"><span class="pre">Predicate&lt;WorkflowContext&gt;</span></code>.
The <code class="docutils literal notranslate"><span class="pre">apply</span></code> method in the predicate takes <code class="docutils literal notranslate"><span class="pre">WorkflowContext</span></code> as a parameter. The Hadoop counters emitted by
the previous MapReduce program (in our case <code class="docutils literal notranslate"><span class="pre">PurchaseEventParser</span></code>) can be retrieved in this method using
the <code class="docutils literal notranslate"><span class="pre">workflowContext</span></code> object. We query for the number of input records to the mappers and the number of records
emitted by the mappers. If at least 80% of the records were successfully parsed and emitted as <code class="docutils literal notranslate"><span class="pre">Purchase</span></code>
by the mappers, the method returns true and the <code class="docutils literal notranslate"><span class="pre">fork</span></code> in the Workflow will be executed. If the method
returns false, the <code class="docutils literal notranslate"><span class="pre">otherwise</span></code> section in the <code class="docutils literal notranslate"><span class="pre">condition</span></code> is executed, which contains the <code class="docutils literal notranslate"><span class="pre">ProblemLogger</span></code>
custom action.</p>
</div>
</div>
<div class="section" id="build-and-run-application">
<h2>Build and Run Application<a class="headerlink" href="#build-and-run-application" title="Permalink to this headline">🔗</a></h2>
<p>The <code class="docutils literal notranslate"><span class="pre">PurchaseWorkflowApp</span></code> can be built and packaged using the Apache Maven command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> mvn clean package
</pre></div>
</div>
<p>Note that the remaining commands assume that the <code class="docutils literal notranslate"><span class="pre">cdap-cli.sh</span></code> script is
available on your PATH. If this is not the case, please add it:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> <span class="nb">export</span> <span class="nv">PATH</span><span class="o">=</span><span class="nv">$PATH</span>:&lt;CDAP home&gt;/bin
</pre></div>
</div>
<p>If you haven’t already started a standalone CDAP installation, start it with the command:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap.sh start
</pre></div>
</div>
<p>We can then deploy the application to the standalone CDAP installation:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh load artifact target/cdap-workflow-guide-&lt;version&gt;.jar
<span class="gp">$</span> cdap-cli.sh create app PurchaseWorkflowApp cdap-workflow-guide &lt;version&gt; user
</pre></div>
</div>
<p>Next, we will send some sample purchase events into the stream
for processing:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh send stream purchaseEvents <span class="s1">&#39;&quot;bob bought 3 apples for $30&quot;&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream purchaseEvents <span class="s1">&#39;&quot;joe bought 1 apple for $100&quot;&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream purchaseEvents <span class="s1">&#39;&quot;joe bought 10 pineapples for $20&quot;&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream purchaseEvents <span class="s1">&#39;&quot;cat bought 3 bottles for $12&quot;&#39;</span>
<span class="gp">$</span> cdap-cli.sh send stream purchaseEvents <span class="s1">&#39;&quot;cat bought 2 pops for $14&quot;&#39;</span>
</pre></div>
</div>
<p>We can now start the Workflow to process the events that were
ingested:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start workflow PurchaseWorkflowApp.PurchaseWorkflow
</pre></div>
</div>
<p>The Workflow will take a couple of minutes to execute.</p>
<p>We can then start the <code class="docutils literal notranslate"><span class="pre">PurchaseResultService</span></code> and query the processed
results:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> cdap-cli.sh start service PurchaseWorkflowApp.PurchaseResultService
</pre></div>
</div>
<ul>
<li><p class="first">Retrieve the purchase records for customer <code class="docutils literal notranslate"><span class="pre">joe</span></code>:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> curl http://localhost:10000/v3/namespaces/default/apps/PurchaseWorkflowApp/services/PurchaseResultService/methods/purchaserecords/joe
</pre></div>
</div>
<p>Example output:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">[{&quot;customer&quot;:&quot;joe&quot;,&quot;product&quot;:&quot;pineapple&quot;,&quot;quantity&quot;:10,&quot;price&quot;:20,&quot;purchaseTime&quot;:1430962917227},{&quot;customer&quot;:&quot;joe&quot;,&quot;product&quot;:&quot;apple&quot;,&quot;quantity&quot;:1,&quot;price&quot;:100,&quot;purchaseTime&quot;:1430962917227}]</span>
</pre></div>
</div>
</li>
<li><p class="first">Retrieve the total purchases made by customer <code class="docutils literal notranslate"><span class="pre">joe</span></code>:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> curl http://localhost:10000/v3/namespaces/default/apps/PurchaseWorkflowApp/services/PurchaseResultService/methods/purchases/customers/joe
</pre></div>
</div>
<p>Example output:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">120</span>
</pre></div>
</div>
</li>
<li><p class="first">Retrieve the total purchases made for product <code class="docutils literal notranslate"><span class="pre">apple</span></code>:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="gp">$</span> curl http://localhost:10000/v3/namespaces/default/apps/PurchaseWorkflowApp/services/PurchaseResultService/methods/purchases/products/apple
</pre></div>
</div>
<p>Example output:</p>
<div class="highlight-console notranslate"><div class="highlight"><pre><span></span><span class="go">130</span>
</pre></div>
</div>
</li>
</ul>
<p>You have now seen how to write a Workflow to connect different MapReduce programs and run them in
parallel based on a condition.</p>
</div>
<div class="section" id="related-topics">
<h2>Related Topics<a class="headerlink" href="#related-topics" title="Permalink to this headline">🔗</a></h2>
<ul class="simple">
<li><a class="reference external" href="https://github.com/cdap-guides/cdap-mapreduce-guide">CDAP MapReduce Guide</a> tutorial for MapReduce</li>
<li><a class="reference external" href="http://docs.cdap.io/cdap/current/en/examples-manual/tutorials/wise.html">Wise: Web Analytics</a> tutorial, part of CDAP</li>
</ul>
</div>
<div class="section" id="share-and-discuss">
<h2>Share and Discuss!<a class="headerlink" href="#share-and-discuss" title="Permalink to this headline">🔗</a></h2>
<p>Have a question? Discuss at the <a class="reference external" href="https://groups.google.com/forum/#!forum/cdap-user">CDAP User Mailing List</a>.</p>
</div>
<div class="section" id="license">
<h2>License<a class="headerlink" href="#license" title="Permalink to this headline">🔗</a></h2>
<p>Copyright © 2015 Cask Data, Inc.</p>
<p>Licensed under the Apache License, Version 2.0 (the “License”); you may
not use this file except in compliance with the License. You may obtain
a copy of the License at</p>
<p><a class="reference external" href="http://www.apache.org/licenses/LICENSE-2.0">http://www.apache.org/licenses/LICENSE-2.0</a></p>
<p>Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an “AS IS” BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.</p>
</div>
</div>

</div>
    <div class="col-md-2">
      <div id="right-sidebar" class="bs-sidenav scrollable-y" role="complementary">
        <div id="localtoc-scrollspy">
        </div>
      </div>
    </div></div>
</div>
<!-- block main content end -->
<!-- block footer -->
<footer class="footer">
      <div class="container">
        <div class="row">
          <div class="col-md-2 footer-left"><a title="Consuming Twitter Data in Real Time" href="cdap-twitter-ingest-guide.html" />Previous</a></div>
          <div class="col-md-8 footer-center"><a class="footer-tab-link" href="../table-of-contents/../../reference-manual/licenses/index.html">Copyright</a> &copy; 2014-2020 Cask Data, Inc.&bull; <a class="footer-tab-link" href="//docs.cask.co/cdap/6.1.1/cdap-docs-6.1.1-web.zip" rel="nofollow">Download</a> an archive or
<a class="footer-tab-link" href="//docs.cask.co/cdap">switch the version</a> of the documentation
          </div>
          <div class="col-md-2 footer-right">&mdash;</div>
        </div>
      </div>
    </footer>
<!-- block footer end -->
<script type="text/javascript" src="../_static/bootstrap-3.3.6/js/bootstrap.min.js"></script><script type="text/javascript" src="../_static/js/bootstrap-sphinx.js"></script><script type="text/javascript" src="../_static/js/abixTreeList-2.js"></script><script type="text/javascript" src="../_static/js/cdap-dynamicscrollspy-4.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script><script type="text/javascript" src="../_static/js/copy-to-clipboard.js"></script><script type="text/javascript" src="../_static/js/jquery.mousewheel.min.js"></script><script type="text/javascript" src="../_static/js/jquery.mCustomScrollbar.js"></script><script type="text/javascript" src="../_static/js/js.cookie.js"></script><script type="text/javascript" src="../_static/js/tabbed-parsed-literal-0.2.js"></script><script type="text/javascript" src="../_static/js/cdap-onload-javascript.js"></script><script type="text/javascript" src="../_static/js/cdap-version-menu.js"></script>
    <script src="https://cdap.gitee.io/docs/cdap/json-versions.js"/></script>
  </body>
</html>